package net.bwie.realtime.vehicle.function;

import net.bwie.realtime.vehicle.bean.CarBean;
import net.bwie.realtime.vehicle.bean.TripDetectionResult;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class JourneyRealtimeFunction extends KeyedProcessFunction<String, CarBean, TripDetectionResult> {
    // 简化的状态定义
    private ValueState<Long> tripStartTimeState;// 记录行程开始时间
    private ValueState<Double> startMileageState;// 记录行程开始里程数

    @Override
    public void open(Configuration conf) {
        // 初始化状态
        tripStartTimeState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("tripStartTime", Long.class));
        startMileageState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("startMileage", Double.class));
    }
    @Override
    public void processElement(CarBean data, Context ctx, Collector<TripDetectionResult> out) throws Exception {
        Long tripStartTime = tripStartTimeState.value();
        Double startMileage = startMileageState.value();
        // 简化逻辑：只要速度>0就认为是行程开始
        if (data.getSpeed() > 0) {
            if (tripStartTime == null) {
                tripStartTimeState.update(data.getTimestamp());
                startMileageState.update(data.getMileage());
            }
        }
        // 简化逻辑：只要速度=0就认为是行程结束
        else {
            if (tripStartTime != null) {
                TripDetectionResult result = new TripDetectionResult();
                // 设置结果中的车架号
                result.setVin(data.getVin());
                // 获取行程ID
                result.setTripId(data.getVin() + "-" + tripStartTime);
                // 获取开始时间
                result.setStartTime(tripStartTime);
                // 结束时间
                result.setEndTime(data.getTimestamp());
                // 计算行驶距离，通过当前里程数减去起始里程数得到
                result.setDistance(data.getMileage() - startMileage);
                out.collect(result);
                // 清除状态
                tripStartTimeState.clear();
                startMileageState.clear();
            }
        }
    }
}
